package bean;

import com.alipay.sofa.jraft.rhea.options.PlacementDriverOptions;
import com.alipay.sofa.jraft.rhea.options.RheaKVStoreOptions;
import com.alipay.sofa.jraft.rhea.options.StoreEngineOptions;
import com.alipay.sofa.jraft.rhea.options.configured.MemoryDBOptionsConfigured;
import com.alipay.sofa.jraft.rhea.options.configured.PlacementDriverOptionsConfigured;
import com.alipay.sofa.jraft.rhea.options.configured.RheaKVStoreOptionsConfigured;
import com.alipay.sofa.jraft.rhea.options.configured.StoreEngineOptionsConfigured;
import com.alipay.sofa.jraft.rhea.storage.StorageType;
import com.alipay.sofa.jraft.util.Endpoint;
import com.alipay.sofa.rpc.config.ConsumerConfig;
import com.alipay.sofa.rpc.listener.ChannelListener;
import com.alipay.sofa.rpc.transport.AbstractChannel;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import io.vertx.core.Vertx;
import io.vertx.core.datagram.DatagramSocket;
import io.vertx.core.datagram.DatagramSocketOptions;
import lombok.*;
import lombok.extern.log4j.Log4j2;
import org.apache.commons.io.FileUtils;
import thirdpart.codec.IBodyCodec;
import thirdpart.fetchserv.IFetchService;

import java.io.File;
import java.io.IOException;
import java.util.Map;
import java.util.Properties;
import java.util.Timer;

@Log4j2
@RequiredArgsConstructor
@ToString
public class SeqConfig {

    private String dataPath;

    private String serverUrl;

    private String serverList;

    private String fetchurls;

    @ToString.Exclude
    @Getter
    private Map<String,IFetchService> fetchServiceMap = Maps.newConcurrentMap();

    @NonNull
    private String fileName;

    @NonNull
    @ToString.Exclude
    @Getter
    private IBodyCodec codec;

    @RequiredArgsConstructor
    private class FetchChannelListener implements ChannelListener{

        @NonNull
        private ConsumerConfig<IFetchService> config;

        @Override
        public void onConnected(AbstractChannel channel) {
            String remoteAddr = channel.remoteAddress().toString();
            log.info("connect to gateway:{}",remoteAddr);
            fetchServiceMap.put(remoteAddr,config.refer());
        }

        @Override
        public void onDisconnected(AbstractChannel channel) {
            String remoteAddr = channel.remoteAddress().toString();
            log.info("disConnect to gateway:{}",remoteAddr);
            fetchServiceMap.remove(remoteAddr);
        }
    }

    public void startup() throws Exception{
        //1 读取配置文件
        initConfig();
        //2 初始化key-value store集群
        startSeqDbCluster();
        //3 启动下游广播
        startMulticast();
        //4 初始化网关链接
        startupFetch();
    }

    @Getter
    private String multicastIp;

    @Getter
    private int multicastPort;

    @Getter
    private DatagramSocket multicastSender;

    private void startMulticast() {
        multicastSender = Vertx.vertx().createDatagramSocket(new DatagramSocketOptions());
    }

    private void startupFetch() {
        //1 建立所有网关的链接
        String[] urls = fetchurls.split(";");
        for (String url : urls) {
            ConsumerConfig<IFetchService> consumerConfig = new ConsumerConfig<IFetchService>()
                    .setInterfaceId(IFetchService.class.getName())
                    .setProtocol("bolt")
                    .setTimeout(5000)
                    .setDirectUrl(url);
            consumerConfig.setOnConnect(Lists.newArrayList(new FetchChannelListener(consumerConfig)));
            fetchServiceMap.put(url,consumerConfig.refer());
        }
        //定时任务 定时去网关抓取数据
        new Timer().schedule(new FetchTask(this),5000,1000);
    }

    @Getter
    private Node node;

    private void startSeqDbCluster() {
        final PlacementDriverOptions pdOpts = PlacementDriverOptionsConfigured.newConfigured()
                .withFake(true).config();
        String[] split = serverUrl.split(":");
        final StoreEngineOptions storeOpts = StoreEngineOptionsConfigured.newConfigured()
                .withStorageType(StorageType.Memory)
                .withMemoryDBOptions(MemoryDBOptionsConfigured.newConfigured().config())
                .withRaftDataPath(dataPath)
                .withServerAddress(new Endpoint(split[0],Integer.parseInt(split[1])))
                .config();

        final RheaKVStoreOptions opts = RheaKVStoreOptionsConfigured.newConfigured()
                .withInitialServerList(serverList)
                .withStoreEngineOptions(storeOpts)
                .withPlacementDriverOptions(pdOpts)
                .config();

        node = new Node(opts);
        node.start();

        //把KVStore的节点的stop方法挂到系统的shutdown()上面去，挂到JDK系统的这个流程当中去
        Runtime.getRuntime().addShutdownHook(new Thread(node::stop));

        log.info("start seq node success on port:{}",split[1] );
    }

    private void initConfig() throws IOException {
        Properties properties = new Properties();
        properties.load(FileUtils.openInputStream(new File("D:\\ideaProject\\matching-trading-system\\seq\\target\\classes\\"+fileName)));
        dataPath=properties.getProperty("datapath");
        serverUrl=properties.getProperty("serverurl");
        serverList=properties.getProperty("serverlist");
        fetchurls=properties.getProperty("fetchurls");

        multicastIp=properties.getProperty("multicastip");
        multicastPort=Integer.parseInt(properties.getProperty("multicastport"));

        log.info("read config:{}", this);
    }


}
